/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.prediction.recset;

import aiaudio.processing.prediction.recset.mappers.CopyAllDataJobMapper;
import aiaudio.processing.prediction.recset.mappers.RecommendationSetExtractorMapper;
import aiaudio.processing.prediction.recset.reducers.CopyAllArtistsJobReducer;
import aiaudio.lastfm.hbase.CannotCreateTableException;
import aiaudio.Application;
import aiaudio.processing.MainTableGroup;
import java.io.IOException;
import java.util.List;
import java.util.Vector;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;

/**
 *
 * @author nastya
 */
public class RecomendationSetCreationAlgorithm {

    private final MainTableGroup mainTableGroup;
    private String temporaryMatrixTable;

    public RecomendationSetCreationAlgorithm(MainTableGroup mainTableGroup) {
        this.mainTableGroup = mainTableGroup;
    }

    public void prepare() throws IOException {
        temporaryMatrixTable = Application.database().createTemporaryMatrixTable();
    }

    public void start() throws IOException, CannotCreateTableException, InterruptedException, ClassNotFoundException {
        List<String[]> userBatches = new Vector<String[]>();
        HTable users = Application.database().getByName(mainTableGroup.getInitialDataTableGroup().getUsersTable()).getTable();


        Scan scan = new Scan();
        ResultScanner scanner = users.getScanner(scan);

        Result[] batch;

        while ((batch = scanner.next(500)) != null && batch.length > 0) {
            String[] userList = new String[batch.length];

            for (int i = 0; i < batch.length; i++) {
                Result result = batch[i];
                byte[] userId = result.getRow();
                String user = Bytes.toString(userId);
                userList[i] = user;
            }
            userBatches.add(userList);
        }


        for (String[] strings : userBatches) {
            Job j = createCopyAllUsresJob(strings);
            j.waitForCompletion(true);
        }


        Job job = createGetNonExistingUsersJob();
        job.waitForCompletion(true);
    }

    protected Job createCopyAllUsresJob(String... users) throws IOException {
        Job job = new Job();

        job.getConfiguration().setStrings(CopyAllArtistsJobReducer.ARTISTS_ARRAY, users);

        Scan scan = new Scan();

        String artistsTableName = Application.database().getByName(mainTableGroup.getInitialDataTableGroup().getArtistsTable()).getDatabaseName();
        TableMapReduceUtil.initTableMapperJob(artistsTableName, scan, CopyAllDataJobMapper.class, ImmutableBytesWritable.class,
                IntWritable.class, job);

        TableMapReduceUtil.initTableReducerJob(temporaryMatrixTable, CopyAllArtistsJobReducer.class, job);

        return job;
    }

    protected Job createGetNonExistingUsersJob() throws IOException {
        Job job = new Job();

        job.getConfiguration().set(RecommendationSetExtractorMapper.COLLECTED_DATA_SET_NAME, mainTableGroup.getSplitDataSetAlgTableGroup().getTrainingRatingMatrix());
        job.getConfiguration().set(RecommendationSetExtractorReducer.INITIAL_DATA_SET_NAME, mainTableGroup.getPrepareRatingsAlgTableGroup().getNormilizedRatingMatrix());

        Scan scan = new Scan();

        TableMapReduceUtil.initTableMapperJob(temporaryMatrixTable, scan, RecommendationSetExtractorMapper.class, ImmutableBytesWritable.class,
                ImmutableBytesWritable.class, job);

        String resultTableName = Application.database().getByName(mainTableGroup.getRatingPredictionAlgTableGroup().getRecommendationCreationRatingList()).getDatabaseName();
        TableMapReduceUtil.initTableReducerJob(resultTableName, RecommendationSetExtractorReducer.class, job);

        return job;
    }
}
